package com.test.source;

import com.alibaba.fastjson.JSON;
import com.test.entry.Order;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.joda.time.DateTime;

import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@PublicEvolving
public class RocketMQConsumer extends RichSourceFunction<String> {

    private static final long serialVersionUID = 1L;
    private String topic;
    private volatile boolean isRunning;
    private DefaultMQPushConsumer consumer;

    public RocketMQConsumer(String topic) {
        this.topic = topic;
        this.isRunning = true;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        Properties properties = getProp();

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(properties.getProperty("group-name"));

        consumer.setConsumeMessageBatchMaxSize(Integer.valueOf(properties.getProperty("consumer.consume-message-batch-max-size")));
        consumer.setMaxReconsumeTimes(Integer.valueOf(properties.getProperty("max-reconsume-times", "16")));
        consumer.setNamesrvAddr(properties.getProperty("rocketmq.name-srv-addr"));

        consumer.setConsumeThreadMin(3);
        consumer.setInstanceName(UUID.randomUUID().toString());
        consumer.subscribe(topic, "*");
        this.consumer = consumer;
    }


    private Properties getProp() {
        Properties properties = new Properties();
        // 使用InPutStream流读取properties文件
        try {
            InputStream in = RocketMQConsumer.class.getClassLoader().getResourceAsStream("rocketMqConsumer.properties");
            properties.load(in);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return properties;
    }

    @Override
    public void run(SourceContext<String> sourceContext) {


        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    try {
                        Order order = JSON.parseObject(new String(messageExt.getBody()), Order.class);
                        System.out.println("收到消息:" + JSON.toJSONString(order) + ":" + new DateTime(order.getCreateDate()).toString("yyyy-MM-dd hh:mm:ss"));
                        sourceContext.collect(new String(messageExt.getBody()));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        try {
            consumer.start();
            while (isRunning) {
                TimeUnit.SECONDS.sleep(100L);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    @Override
    public void cancel() {
        System.out.println("close 消费者");
        if (null != consumer) {
            consumer.shutdown();
        }
        isRunning = false;
    }
}
